package org.example.watemark;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.time.Duration;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 15:53
 */
public class WindowWaterMakeIdleness {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        environment.setParallelism(2);

        SingleOutputStreamOperator<Integer> streamOperator =
                environment.socketTextStream("127.0.0.1", 7777)
                        .map(Integer::parseInt)
                        .partitionCustom(new Partitioner<Integer>() {
                            @Override
                            public int partition(Integer key, int numPartitions) {
                                return key % numPartitions;
                            }
                        }, (KeySelector<Integer, Integer>) value -> value)
                        .assignTimestampsAndWatermarks(
                                WatermarkStrategy.<Integer>forMonotonousTimestamps()
                                        .withTimestampAssigner((TimestampAssignerSupplier<Integer>) t ->
                                                        (TimestampAssigner<Integer>) (element, recordTimestamp) -> {
                                                            System.out.println("element---> " + element);
                                                            return element * 1000L;
                                                        }
                                        ).withIdleness(Duration.ofSeconds(2))

                        )
                ;

        streamOperator
                .keyBy(v -> v % 2)
                // 窗口 大小 5秒
                .window(TumblingEventTimeWindows.of(Time.seconds(8)))
                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, ProcessWindowFunction<Integer, String,
                            Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                        String startTime = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss");
                        String endTime = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss");
                        out.collect(
                                "窗口开始时间：" + startTime +
                                        "窗口结束时间：" + endTime +
                                        " 窗口内数据条数：" + elements.spliterator().estimateSize() + " 窗口内数据：" + elements);
                    }
                })
                .print();

        environment.execute();
    }

}
